package datastore.api.impl.hbase;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Scanner;
import org.apache.hadoop.hbase.filter.RegExpRowFilter;
import org.apache.hadoop.hbase.io.BatchUpdate;
import org.apache.hadoop.hbase.util.Bytes;

import datastore.db.Model;
import datastore.db.BaseQuery;

public class KindIdxStore extends BaseStore<Scanner, BaseQuery<?>> {
    private static final String TABLE_NAME = "kind_idx";
    private static final String COL_FAM = "kind:";
    private static final String COL_QUALIFIER = "value";
    private static final String COLUMN = COL_FAM + COL_QUALIFIER;

    public KindIdxStore(HBaseDatastore hbase) {
        super(hbase);
    }
    
    @Override
    public void init() throws Exception {
        List<BatchUpdate> updates = new ArrayList<BatchUpdate>();
        
        //Create table
        boolean tableExists = admin.tableExists(TABLE_NAME);
        if (!tableExists) {
            createTable(updates, TABLE_NAME, COL_FAM, COLUMN, 
                    constructKey(DEFAULT_KEY_PART, DEFAULT_KEY_PART));
        }

        table = new HTable(config, TABLE_NAME);
        /*for (String kind : kinds) {
            String rowKey = constructKey(kind, DEFAULT_KEY_PART);
            if (!tableExists || !table.exists(rowKey.getBytes())) {
                updates.add(getBatch(rowKey, COLUMN, EMPTY_VALUE));
            }
        }*/
        if (!updates.isEmpty()) {
            table.commit(updates);
        }
    }

    @Override
    public void put(Model model) throws Exception {
        putRow(toRowKey(model), TABLE_NAME, COLUMN, EMPTY_VALUE);
    }

    @Override
    public void delete(Model model) throws Exception {
        deleteRow(toRowKey(model), TABLE_NAME);
    }
    
    @Override
    public Scanner get(BaseQuery<?> query) throws Exception {
        String kind = query.getModelClass().getName();
        String regex = appName + Store.KEY_DELIM + kind + Store.KEY_DELIM+".*";
        String startRow = constructKey(kind, DEFAULT_KEY_PART);
        Scanner scanner = table.getScanner(new byte[][] { Bytes.toBytes(COLUMN)}, 
                Bytes.toBytes(startRow), new RegExpRowFilter(regex));
        return scanner;
    }
    
    protected String toRowKey(Model model) {
        String entityRowKey = escape(toEntityRowKey(model.getKey()));
        String rowKey = constructKey(model.getClass().getName(), entityRowKey);
        return rowKey;
    }

}
